/*
 * Copyright (c) Facebook, Inc. and its affiliates.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#pragma once

#include "velox/common/base/BitUtil.h"
#include "velox/common/base/Exceptions.h"
#include "DeltaBpDecoder.h"

namespace facebook::velox::parquet {

// DeltaBpDecoder is adapted from Apache Arrow:
// https://github.com/apache/arrow/blob/apache-arrow-12.0.0/cpp/src/parquet/encoding.cc#LL2357C18-L2586C3
//template <typename T>
class DeltaLengthBaDecoder {
 public:
  explicit DeltaLengthBaDecoder(const char* start, int32_t encode_size) 
	               :bufferStart_(start),
		        page_size_(encode_size),
                  len_decoder_(start){
    std::cout << "@@ init deltaLenBaDecoder \n";

    DecodeLengths();
  }

  explicit DeltaLengthBaDecoder(){}

  void SetDecode(const char* start, int32_t encode_size, uint32_t read_nums = 1){
    bufferStart_ = start;
    page_size_ = encode_size;
    len_decoder_.SetDecode(start, read_nums);
    std::cout << "@@ SetDecode deltaLenBaDecoder \n";
    DecodeLengths();
  }

  void skip(uint64_t numValues) {
    skip<false>(numValues, 0, nullptr);
  }

  template <bool hasNulls>
  inline void skip(int32_t numValues, int32_t current, const uint64_t* nulls) {
    if (hasNulls) {
      numValues = bits::countNonNulls(nulls, current, current + numValues);
    }
    for (int32_t i = 0; i < numValues; ++i) {
      readString();
    }
  }
/*
  template <bool hasNulls, typename Visitor>
  void readWithVisitor(const uint64_t* nulls, Visitor visitor) {
    std::cout << "@@ readWithVisitor deltaBaDecoder \n";
    int32_t current = visitor.start();
    skip<hasNulls>(current, 0, nulls);
    int32_t toSkip;
    bool atEnd = false;
    const bool allowNulls = hasNulls && visitor.allowNulls();
    for (;;) {
      if (hasNulls && allowNulls && bits::isBitNull(nulls, current)) {
        toSkip = visitor.processNull(atEnd);
      } else {
        if (hasNulls && !allowNulls) {
          toSkip = visitor.checkAndSkipNulls(nulls, current, atEnd);
          if (!Visitor::dense) {
            skip<false>(toSkip, current, nullptr);
          }
          if (atEnd) {
            return;
          }
        }

        // We are at a non-null value on a row to visit.
        toSkip = visitor.process(readString(), atEnd);
      }
      ++current;
      if (toSkip) {
        skip<hasNulls>(toSkip, current, nulls);
        current += toSkip;
      }
      if (atEnd) {
        return;
      }
    }
  }
*/
  std::string_view getString(){
    return readString();
  }

 private:

  bool Advance(int64_t num_bits){
    int64_t bits_required = bitOffset_ + num_bits;
    int64_t bytes_required = bits::nbytes(bits_required);
    if(bytes_required > (page_size_ - byteOffset_)){
      return false;
    }

    byteOffset_ += bytes_required;
    bitOffset_ = bits_required & 0x7;

    return true;
  }

  std::string_view  readString() {
    std::cout << "@@ readString deltalengthBaDecoder \n";
    //std::string_view value;
    int64_t data_size = 0;
    int64_t byteOffset = byteOffset_; 
    int64_t v_len = buffered_length_[length_idx_];
    if(v_len < 0){
      VELOX_FAIL("negative string delta length");
    }
    length_idx_++;
    data_size += v_len;

    if(!Advance(8*data_size)){
      VELOX_FAIL("advance error");
    }
    const char* data_ptr = bufferStart_ + byteOffset;

    std::string_view value(data_ptr, v_len);
    lastValue_  = value;
    //value = lastValue_;
    bufferStart_ += v_len;

    numValidValues_ --;
    
    return value;
  }

/*
  uint32_t calcByteOffset(const char* org, const char* cur){
    return uint32_t(cur - org);
  }
*/
  void DecodeLengths() {
    // get the number of encoded lengths
    int num_length = len_decoder_.ValidValuesCount();
    for(int i = 0; i < num_length; i++){
      uint64_t v = len_decoder_.getReadLong();
      buffered_length_.push_back(v);
    }
    uint32_t byteOffset= 0;
    uint32_t bitOffset = 0;
    len_decoder_.getOffset(byteOffset, bitOffset);
    
    byteOffset_ = byteOffset;
    bitOffset_ = bitOffset;

    length_idx_ = 0;
    numValidValues_ = num_length;
  }

  const char* bufferStart_;
  int32_t page_size_;
  uint32_t length_idx_{0};
  std::string_view lastValue_;
  int64_t bitOffset_{0};
  int64_t byteOffset_{0};
  int numValidValues_{0};
  uint32_t prefixLenOffset_{0};
  std::vector<uint64_t> buffered_length_;


  DeltaBpDecoder len_decoder_;
};

} // namespace facebook::velox::parquet
